[FLINK-40113][checkpoint] Reject restore when operators on a keyed vertex disagree on max parallelism by rionmonster · Pull Request #28714 · apache/flink

rionmonster · 2026-07-10T19:21:04Z

What is the purpose of the change

When operators are chained into a single vertex, each operator's recorded maximum parallelism is reconciled onto that shared vertex individually during state restore. Since FLINK-31996 relaxed chaining to allow operators with differing maximum parallelism to share a vertex, a keyed operator can be silently restored under a key-group count that is not its own if the chaining topology has changed since the checkpoint — its keyed state is then remapped through an incompatible hash % maxParallelism layout.

Flink already rejects a single operator's incompatible maximum-parallelism change on restore, but never validates that operators sharing a vertex agree with each other, so the per-vertex reconciliation silently adopts whichever value is applied last. This change closes that consistency gap by rejecting such a restore with a clear error instead of restoring keyed state improperly.

Brief change log

Add checkMaxParallelismAgreement in StateAssignmentOperation, invoked before the per-operator reconciliation in checkParallelismPreconditions.
On a vertex that carries keyed state, fail with a clear IllegalStateException when its operators recorded differing maximum parallelism; the comparison spans all chained operators, since a non-keyed operator's recorded value can win the reconciliation and misroute a keyed operator's state.
Add ChainingMaxParallelismStateLossITCase covering both directions (explicit value below and above the chain head's) on the HashMap and RocksDB backends.

Verifying this change

This change added tests and can be verified as follows:

ChainingMaxParallelismStateLossITCase takes a canonical savepoint with operator chaining disabled (the keyed operator is its own vertex at an explicit maximum parallelism), then restores with chaining enabled (the keyed operator chains under the source head, whose derived maximum parallelism differs). It asserts the restore is rejected with the new error for both directions and both state backends.
Existing StateAssignmentOperationTest and CheckpointCoordinatorRestoringTest continue to pass, confirming the new check is additive and runs only on restore.

Does this pull request potentially affect one of the following parts:

Dependencies (does it add or upgrade a dependency): no
The public API, i.e., is any changed class annotated with @Public(Evolving): no
The serializers: no
The runtime per-record code paths (performance sensitive): no
Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: yes — adds a restore-time precondition in checkpoint state assignment, executed once per vertex during recovery.
The S3 file system connector: no

Documentation

Does this pull request introduce a new feature? no
If yes, how is the feature documented? not applicable

Was generative AI tooling used to co-author this PR?

Yes (please specify the tool below)

Generated-by: Claude Opus 4.8 (1M context)

…ined operators disagree on max parallelism

…rtex disagree on max parallelism

flinkbot · 2026-07-10T19:30:05Z

CI report:

69d2d41 Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run azure re-run the last Azure build

pnowojski · 2026-07-13T15:26:36Z

+        // reinterpretAsKeyedStream puts a forward (chainable) edge before the keyed operator, so it
+        // can become a chained non-head under the source head.
+        final KeyedStream<Integer, Integer> keyed =
+                DataStreamUtils.reinterpretAsKeyedStream(
+                        source, (KeySelector<Integer, Integer>) value -> value % NUM_KEYS);


Can this issue be reproduced without reinterpretAsKeyedStream? 🤔 Just normal keyBy followed by a chained/not chained operator?

reinterpretAsKeyedStream is a bit hacky and doesn't represents the normal use cases. It's intention is if partitions in your sources are already keyed by something, so you don't need to keyBy that data stream again. Here it only works because you constrain the parallelism to 1 which is also a bit artificial constraint.

Yes, I agree and don't typically don't prefer using it.

I went down that road and tried iterating several times with just a plain keyBy to trigger the issue with very no luck. Per Claude:

keyBy inserts a hash (KeyGroupStreamPartitioner) edge, which breaks the chain, so the keyed operator always ends up as a chain head with its own maxParallelism. There's no second operator on the vertex to disagree with, and on restore it just adopts its own recorded value. To hit this the keyed operator has to be a chained non-head sharing a vertex with something that recorded a different maxParallelism, and reinterpretAsKeyedStream (forward edge) is the only pure-streaming way to get that topology.

I'll try exploring a few other ways to reproduce this issue without it and report back. 👍

pnowojski · 2026-07-13T15:32:19Z

+    private void startCluster(String backend) throws Exception {
+        final Configuration config = new Configuration();
+        config.set(StateBackendOptions.STATE_BACKEND, backend);
+        config.set(
+                CheckpointingOptions.CHECKPOINTS_DIRECTORY,
+                TempDirUtils.newFolder(temporaryFolder).toURI().toString());
+        config.set(
+                CheckpointingOptions.SAVEPOINT_DIRECTORY,
+                TempDirUtils.newFolder(temporaryFolder).toURI().toString());
+        // Default is already true; set explicitly for clarity — this is what lets the keyed
+        // operator
+        // chain under a head with a different (auto-derived) max parallelism.
+        config.set(
+                PipelineOptions.OPERATOR_CHAINING_CHAIN_OPERATORS_WITH_DIFFERENT_MAX_PARALLELISM,
+                true);
+
+        cluster =
+                new MiniClusterWithClientResource(
+                        new MiniClusterResourceConfiguration.Builder()
+                                .setConfiguration(config)
+                                .setNumberTaskManagers(1)
+                                .setNumberSlotsPerTaskManager(4)
+                                .build());
+        cluster.before();
+    }


Why do you need this manually started instead of using @ClassRule on the MiniClusterWithClientResource field? There are plenty of examples when such cluster has a custom configuration.

Yup, this is fair. I can move it into the job configuration instead. 👍

pnowojski · 2026-07-13T15:37:54Z

+        COUNTS.clear();
+        RESTORE_EFFECTIVE_MAX_PARALLELISM.set(-1);
+
+        final Deadline deadline = Deadline.now().plus(Duration.ofMinutes(2));


Why do we need a deadline? Our default convetion is to just wait indefinetly:

often even reasonable deadlines are violated due to slow CI

if something indeed deadlocks in CI, we have a global timeout of ~4h, after which CI captures thread dump. By adding custom deadlines/timeouts, you are basically bypassing that thread dump capture mechanism, which often is the only way one can analyse a deadlock (especially painful if it's extremely rare and can't be reproduced locally)

I can go ahead and drop this as well to clean things up.

pnowojski · 2026-07-13T16:01:30Z

+ * Verifies that restoring a savepoint is rejected when a chaining change places operators with
+ * different recorded max parallelism onto a single keyed vertex.


Some commits needs to be squashed

Will do! I'll squash it down to a single primary commit.

rionmonster added 2 commits July 10, 2026 14:18

[FLINK-40113][tests] Add ITCase reproducing keyed-state loss when cha…

0fbbcd2

…ined operators disagree on max parallelism

[FLINK-40113][checkpoint] Reject restore when operators on a keyed ve…

69d2d41

…rtex disagree on max parallelism

pnowojski requested changes Jul 13, 2026

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

[FLINK-40113][checkpoint] Reject restore when operators on a keyed vertex disagree on max parallelism#28714

[FLINK-40113][checkpoint] Reject restore when operators on a keyed vertex disagree on max parallelism#28714
rionmonster wants to merge 2 commits into
apache:masterfrom
rionmonster:FLINK-40113

rionmonster commented Jul 10, 2026 •

edited

Loading

Uh oh!

flinkbot commented Jul 10, 2026 •

edited

Loading

Uh oh!

pnowojski Jul 13, 2026

Uh oh!

rionmonster Jul 13, 2026

Uh oh!

pnowojski Jul 13, 2026

Uh oh!

rionmonster Jul 13, 2026

Uh oh!

pnowojski Jul 13, 2026

Uh oh!

rionmonster Jul 13, 2026

Uh oh!

pnowojski Jul 13, 2026

Uh oh!

rionmonster Jul 13, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

		* Verifies that restoring a savepoint is rejected when a chaining change places operators with
		* different recorded max parallelism onto a single keyed vertex.

Uh oh!

Conversation

rionmonster commented Jul 10, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What is the purpose of the change

Brief change log

Verifying this change

Does this pull request potentially affect one of the following parts:

Documentation

Was generative AI tooling used to co-author this PR?

Uh oh!

flinkbot commented Jul 10, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

CI report:

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

rionmonster commented Jul 10, 2026 •

edited

Loading

flinkbot commented Jul 10, 2026 •

edited

Loading